package com.sky.demo.redis;


import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.junit.Before;
import org.junit.Test;
import redis.clients.jedis.Jedis;

import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

@Slf4j
@Data
public class RedisUtil {

    private Jedis jedis;
    private static final String LOCK_KEY = "redis_lock";
    private static final String QUEUE_KEY = "redis_queue";
    private final int MAX_POOL_SIZE = 80;
    private final LinkedBlockingDeque<Runnable> consumeQueue = new LinkedBlockingDeque<Runnable>();
    private final LinkedBlockingDeque<Runnable> productQueue = new LinkedBlockingDeque<Runnable>();
    private final ThreadPoolExecutor productExecutor = new ThreadPoolExecutor(10, MAX_POOL_SIZE, 60, TimeUnit.SECONDS, productQueue);
    private final ThreadPoolExecutor consumeExecutor = new ThreadPoolExecutor(10, MAX_POOL_SIZE, 60, TimeUnit.SECONDS, consumeQueue);


    public String set(String key, String value){
        return jedis.set(key, value);
    }

    public String get(String key){
        return jedis.get(key);
    }

    public void lpush(String key, String value){
        jedis.lpush(key, value);
    }

    public String rpop(String key){
        return jedis.rpop(key);
    }

    @Before
    public void setJedis(){
        jedis = new Jedis("127.0.0.1", 6379);
    }


    public RedisUtil(){}

    public RedisUtil(Jedis jedis){
        this.jedis = jedis;
    }

    @Test
    public void test2(){
        jedis.set("name", "g");
        String name = jedis.get("name");
        System.out.println(name);
    }


    @Test
    public void test(){
        for(int i = 0; i < 10; i++){
            jedis.lpush(QUEUE_KEY, "1");
            productExecutor.submit(this::consume);
        }
    }

    private void consume(){
        if(lock(LOCK_KEY, 6000L)){
            log.error("redis 已锁定");
            return;
        }
        try {
            while(true){
                if(consumeExecutor.getActiveCount() >= MAX_POOL_SIZE){
                    Thread.sleep(100);
                }
                String rpop = jedis.rpop(QUEUE_KEY);
                if(rpop == null){
                    break;
                }
                consumeExecutor.submit(()->{
                    log.info("模拟业务处理");
                    try {
                        Thread.sleep(5000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                });
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }finally {
            del(LOCK_KEY);
        }
    }


    private synchronized boolean lock(String key, Long seconds){
        long flag = jedis.setnx(key, key);
        if(flag > 0){
            jedis.expire(key, seconds);
            return true;
        }
        return false;
    }

    private synchronized void del(String key){
        jedis.del(key);
    }
}
